package com.by.common.redis;

import jakarta.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Component;

import java.util.BitSet;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

/**
 * @author lby
 * @description 秒杀系统商品ID校验布隆过滤器
 * @date 2025/8/7 21:19
 */
@Component
public class RedisBloomFilter {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 布隆过滤器参数，根据实际商品数量调整
    private static final long DEFAULT_SIZE = 10_0000 * 14; // 10万商品 * 14倍 = 140万位(约170KB)
    private static final int[] SEEDS = new int[]{3, 5, 7, 11, 13, 31, 37, 61, 73, 97}; // 10个哈希函数
    private static final int BATCH_SIZE = 1024 * 1024; // 每次批量读取1MB数据(800万个bit)

    private final Logger logger = Logger.getLogger(RedisBloomFilter.class.getName());
    private final ConcurrentHashMap<String, BitSet> localCache = new ConcurrentHashMap<>();
    // 线程池用于并行加载数据
    private final ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    /**
     * 初始化方法，应用启动时加载所有布隆过滤器数据
     */

    /**
     * 初始化布隆过滤器（Redis和本地缓存）
     * 供订单模块上传商品后调用
     */
    public void initBoomFilter(String key, List<Long> ids) {
        long startTime = System.currentTimeMillis();

        // 1. 批量添加到Redis
        batchAddToRedis(key, ids);

        // 2. 重新从Redis加载到本地缓存
        loadFromRedis(key);

        logger.info("初始化布隆过滤器完成，key: " + key + ", 商品数量: " + ids.size() +
                ", 耗时: " + (System.currentTimeMillis() - startTime) + "ms");
    }

    /**
     * 添加元素到布隆过滤器（同时更新Redis和本地缓存）
     */
    public void add(String key, String value) {
        addToRedis(key, value);
        addToLocalCache(key, value);
    }

    public void add(String key, Long value) {
        add(key, String.valueOf(value));
    }

    /**
     * 判断元素是否可能存在（优先检查本地缓存）
     */
    public boolean mightContain(String key, String value) {
        BitSet localBitSet = localCache.get(key);
        if (localBitSet == null) {
            // 本地缓存未初始化，先加载
            loadFromRedis(key);
            localBitSet = localCache.get(key);
            // 如果仍然为null，说明不存在
            if (localBitSet == null) {
                return false;
            }
        }

        for (int seed : SEEDS) {
            long hashValue = hash(value, seed);
            if (!localBitSet.get((int) (hashValue % DEFAULT_SIZE))) {
                return false;
            }
        }
        return true;
    }

    public boolean mightContain(String key, Long value) {
        return mightContain(key, String.valueOf(value));
    }

    /**
     * 从Redis一次性加载数据到本地缓存
     */
    public void loadFromRedis(String key) {
        long startTime = System.currentTimeMillis();
        BitSet bitSet = new BitSet((int) DEFAULT_SIZE);

        // 使用CountDownLatch实现多线程并行加载
        CountDownLatch latch = new CountDownLatch(SEEDS.length);

        for (int seed : SEEDS) {
            final int currentSeed = seed;
            executor.submit(() -> {
                try {
                    String hashKey = key + ":hash:" + currentSeed;
                    loadBitsBatch(hashKey, bitSet, currentSeed);
                } finally {
                    latch.countDown();
                }
            });
        }

        try {
            // 等待所有种子的哈希数据加载完成
            latch.await();
            localCache.put(key, bitSet);
            logger.info("从Redis加载布隆过滤器数据完成，key: " + key +
                    ", 耗时: " + (System.currentTimeMillis() - startTime) + "ms");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.warning("加载布隆过滤器数据被中断，key: " + key + ", 错误: " + e.getMessage());
        }
    }

    /**
     * 批量加载Redis中的位数据到本地BitSet
     */
    private void loadBitsBatch(String hashKey, BitSet bitSet, int seed) {
        // 使用Redis的GETRANGE命令批量获取字节数据
        byte[] bytes = redisTemplate.execute((RedisCallback<byte[]>) connection -> {
            byte[] keyBytes = redisTemplate.getStringSerializer().serialize(hashKey);
            if (keyBytes == null) {
                return new byte[0];
            }

            // 计算需要读取的字节数
            long byteLength = (DEFAULT_SIZE + 7) / 8; // 向上取整
            return connection.getRange(keyBytes, 0, byteLength - 1);
        });

        if (bytes != null && bytes.length > 0) {
            // 将字节数组转换为BitSet
            for (int i = 0; i < bytes.length; i++) {
                byte b = bytes[i];
                // 处理每个字节的8个位
                for (int j = 0; j < 8; j++) {
                    long position = (long) i * 8 + j;
                    if (position >= DEFAULT_SIZE) {
                        break;
                    }
                    // 检查第j位是否为1
                    if ((b & (1 << j)) != 0) {
                        bitSet.set((int) position);
                    }
                }
            }
        }
    }

    /**
     * 批量添加多个ID到Redis
     */
    private void batchAddToRedis(String key, List<Long> ids) {
        // 使用管道批量执行命令，减少网络交互
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (Long id : ids) {
                String value = String.valueOf(id);
                for (int seed : SEEDS) {
                    String hashKey = key + ":hash:" + seed;
                    long hashValue = hash(value, seed);
                    long position = hashValue % DEFAULT_SIZE;

                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(hashKey);
                    if (keyBytes != null) {
                        connection.setBit(keyBytes, position, true);
                    }
                }
            }
            return null;
        });
    }

    /**
     * 添加单个元素到Redis
     */
    private void addToRedis(String key, String value) {
        for (int seed : SEEDS) {
            String hashKey = key + ":hash:" + seed;
            long hashValue = hash(value, seed);
            redisTemplate.opsForValue().setBit(hashKey, hashValue % DEFAULT_SIZE, true);
            // 设置过期时间
            redisTemplate.expire(hashKey, 86400, java.util.concurrent.TimeUnit.SECONDS);
        }
    }

    /**
     * 添加元素到本地缓存
     */
    private void addToLocalCache(String key, String value) {
        BitSet bitSet = localCache.computeIfAbsent(key, k -> new BitSet((int) DEFAULT_SIZE));
        for (int seed : SEEDS) {
            long hashValue = hash(value, seed);
            bitSet.set((int) (hashValue % DEFAULT_SIZE));
        }
    }

    /**
     * 高效哈希函数
     */
    private long hash(String value, int seed) {
        long result = 0;
        for (char c : value.toCharArray()) {
            result = seed * result + c;
        }
        return result & 0x7FFFFFFF; // 确保非负
    }
}
